package a.cb;

import org.apache.flink.connector.datagen.table.DataGenConnectorOptions;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.*;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * 创建TableEnvironment
 */
public class MyCreateTableApp {

    public static void main(String[] args) {
//        掺杂DataStreamAPI方式
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        TableEnvironment tableEnvironment = StreamTableEnvironment.create(env);
        tableEnvironment.createTemporaryTable("SourceTable", TableDescriptor.forConnector("datagen")
                .schema(Schema.newBuilder()
                        .column("f0", DataTypes.STRING())
                        .build())
                .option(DataGenConnectorOptions.ROWS_PER_SECOND, 10l)
                .build());

        Table table = tableEnvironment.from("SourceTable");
        table.execute().print();

    }

    /**
     * 创建TableEnvironment
     *
     * @param args
     */
    public static void main1(String[] args) {
        EnvironmentSettings settings = EnvironmentSettings.newInstance()
                .inStreamingMode()
                .build();
        TableEnvironment environment = TableEnvironment.create(settings);
        environment.getConfig().getConfiguration().setString("parallelism.default", "1");
//        读取数据源
//
        environment.createTemporaryTable("SourceTable", TableDescriptor.forConnector("datagen")
                .schema(Schema.newBuilder()
                        .column("f0", DataTypes.STRING())
                        .build())
                .option(DataGenConnectorOptions.ROWS_PER_SECOND, 10l)
                .build());
        Table table = environment.from("SourceTable");
        table.execute().print();


    }
}
